package com.shujia.flink.core

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.base.DeliveryGuarantee
import org.apache.flink.connector.kafka.sink.{KafkaRecordSerializationSchema, KafkaSink}
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import java.time.Duration
import java.util.Date

object Demo9RoadFlow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    /**
     * 1、从kafka中实时读取卡口过车数据
     */
    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      //kafka 集群列表
      .setBootstrapServers("master:9092,node1:9092,node2:9092")
      //消费的topic
      .setTopics("cars")
      //消费者组
      .setGroupId("my-group")
      //读取数据的位置，earliest：从最早读取数据，latest：读取最新数据
      .setStartingOffsets(OffsetsInitializer.earliest)
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build

    //使用kafka 数据源
    val carsDS: DataStream[String] = env
      .fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")


    /**
     * 2、解析数据，取出道路编号和事件时间
     *
     */
    val roadAndTimeDS: DataStream[(Long, Long)] = carsDS.map(car => {
      val carObj: JSONObject = JSON.parseObject(car)
      val road_id: Long = carObj.getLong("road_id")
      val time: Long = carObj.getLong("time") * 1000
      (road_id, time)
    })

    /**
     * 3、指定事件时间和水位线
     */

    val ws: WatermarkStrategy[(Long, Long)] = WatermarkStrategy
      .forBoundedOutOfOrderness[(Long, Long)](Duration.ofSeconds(5))
      .withTimestampAssigner(new SerializableTimestampAssigner[(Long, Long)] {
        override def extractTimestamp(element: (Long, Long), recordTimestamp: Long): Long = element._2
      })

    val assDS: DataStream[(Long, Long)] = roadAndTimeDS.assignTimestampsAndWatermarks(ws)

    /**
     * 4、统计道路车流量
     */
    //安装道路分组
    val keyByDS: KeyedStream[(Long, Long), Long] = assDS.keyBy(_._1)

    //划分窗口
    val windowDS: WindowedStream[(Long, Long), Long, TimeWindow] = keyByDS
      .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(1)))

    //统计车流量
    val roadFlowDS: DataStream[String] = windowDS
      .process(new ProcessWindowFunction[(Long, Long), String, Long, TimeWindow] {
        override def process(roadId: Long,
                             context: Context,
                             elements: Iterable[(Long, Long)],
                             out: Collector[String]): Unit = {
          //车流量
          val flow: Long = elements.size

          //获取窗口结束时间
          val endTime: Long = context.window.getEnd
          val date = new Date(endTime)
          val format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
          val endDate: String = format.format(date)

          //将数据发生到下游
          out.collect(s"$roadId\t$endDate\t$flow")

        }
      })


    /**
     * 将计算结果保存到kafka中
     */

    val kafkaSink: KafkaSink[String] = KafkaSink
      .builder[String]()
      //kafka 集群列表
      .setBootstrapServers("master:9092,node1:9092,node2:9092")
      .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        //指定topic
        .setTopic("road_flow")
        //指定序列化方式
        .setValueSerializationSchema(new SimpleStringSchema())
        .build()
      )
      //指定数据处理的语义AT_LEAST_ONCE：至少一次,EXACTLY_ONCE:唯一一次
      .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
      .build()

    roadFlowDS.sinkTo(kafkaSink)

    env.execute()


  }

}
